//! 共识算法模块
//! 
//! 实现Raft共识算法，确保集群中节点的数据一致性和领导者选举。

use crate::{Error, Result};
use super::{ClusterConfig, ClusterEvent};
use std::collections::HashMap;
use std::time::{SystemTime, Duration};
use serde::{Serialize, Deserialize};
use tokio::sync::mpsc;

/// 共识引擎
#[derive(Debug)]
pub struct ConsensusEngine {
    /// 集群配置
    config: ClusterConfig,
    /// 当前状态
    state: ConsensusState,
    /// 当前任期
    current_term: u64,
    /// 投票给的候选者
    voted_for: Option<String>,
    /// 日志条目
    log: Vec<LogEntry>,
    /// 已提交的日志索引
    commit_index: u64,
    /// 已应用的日志索引
    last_applied: u64,
    /// 选举超时时间
    election_timeout: SystemTime,
    /// 事件发送器
    event_sender: Option<mpsc::UnboundedSender<ClusterEvent>>,
}

/// 共识状态
#[derive(Debug, Clone, PartialEq)]
pub enum ConsensusState {
    /// 跟随者
    Follower,
    /// 候选者
    Candidate,
    /// 领导者
    Leader,
}

/// 日志条目
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LogEntry {
    /// 任期
    pub term: u64,
    /// 索引
    pub index: u64,
    /// 命令
    pub command: Command,
    /// 时间戳
    pub timestamp: SystemTime,
}

/// 命令类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum Command {
    /// 无操作
    NoOp,
    /// 数据操作
    DataOperation {
        operation_type: String,
        key: String,
        value: Option<Vec<u8>>,
    },
    /// 配置变更
    ConfigChange {
        change_type: String,
        node_id: String,
        address: String,
    },
}

/// 投票请求
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoteRequest {
    /// 任期
    pub term: u64,
    /// 候选者ID
    pub candidate_id: String,
    /// 最后日志索引
    pub last_log_index: u64,
    /// 最后日志任期
    pub last_log_term: u64,
}

/// 投票响应
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoteResponse {
    /// 任期
    pub term: u64,
    /// 是否投票
    pub vote_granted: bool,
}

/// 追加条目请求
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AppendEntriesRequest {
    /// 任期
    pub term: u64,
    /// 领导者ID
    pub leader_id: String,
    /// 前一个日志索引
    pub prev_log_index: u64,
    /// 前一个日志任期
    pub prev_log_term: u64,
    /// 日志条目
    pub entries: Vec<LogEntry>,
    /// 领导者提交索引
    pub leader_commit: u64,
}

/// 追加条目响应
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AppendEntriesResponse {
    /// 任期
    pub term: u64,
    /// 是否成功
    pub success: bool,
}

impl ConsensusEngine {
    /// 创建新的共识引擎
    pub fn new(config: ClusterConfig) -> Result<Self> {
        Ok(Self {
            config,
            state: ConsensusState::Follower,
            current_term: 0,
            voted_for: None,
            log: Vec::new(),
            commit_index: 0,
            last_applied: 0,
            election_timeout: SystemTime::now() + Duration::from_millis(5000),
            event_sender: None,
        })
    }
    
    /// 启动共识引擎
    pub async fn start(&mut self) -> Result<()> {
        tracing::info!("启动共识引擎");
        
        // 重置选举超时
        self.reset_election_timeout();
        
        tracing::info!("共识引擎启动完成");
        Ok(())
    }
    
    /// 停止共识引擎
    pub async fn stop(&mut self) -> Result<()> {
        tracing::info!("停止共识引擎");
        Ok(())
    }
    
    /// 开始选举
    pub async fn start_election(&mut self) -> Result<()> {
        tracing::info!("开始领导者选举");
        
        // 转换为候选者状态
        self.state = ConsensusState::Candidate;
        self.current_term += 1;
        self.voted_for = Some(self.config.node_id.clone());
        self.reset_election_timeout();
        
        // 发送投票请求
        self.send_vote_requests().await?;
        
        Ok(())
    }
    
    /// 发送投票请求
    async fn send_vote_requests(&self) -> Result<()> {
        let last_log_index = self.log.len() as u64;
        let last_log_term = self.log.last().map(|entry| entry.term).unwrap_or(0);
        
        let vote_request = VoteRequest {
            term: self.current_term,
            candidate_id: self.config.node_id.clone(),
            last_log_index,
            last_log_term,
        };
        
        tracing::debug!("发送投票请求: {:?}", vote_request);
        
        // 简化实现：实际需要向其他节点发送请求
        
        Ok(())
    }
    
    /// 处理投票请求
    pub async fn handle_vote_request(&mut self, request: VoteRequest) -> Result<VoteResponse> {
        tracing::debug!("处理投票请求: {:?}", request);
        
        let mut vote_granted = false;
        
        // 检查任期
        if request.term > self.current_term {
            self.current_term = request.term;
            self.voted_for = None;
            self.state = ConsensusState::Follower;
        }
        
        // 检查是否可以投票
        if request.term == self.current_term &&
           (self.voted_for.is_none() || self.voted_for.as_ref() == Some(&request.candidate_id)) {
            
            // 检查日志是否至少和当前节点一样新
            let last_log_index = self.log.len() as u64;
            let last_log_term = self.log.last().map(|entry| entry.term).unwrap_or(0);
            
            if request.last_log_term > last_log_term ||
               (request.last_log_term == last_log_term && request.last_log_index >= last_log_index) {
                vote_granted = true;
                self.voted_for = Some(request.candidate_id.clone());
                self.reset_election_timeout();
            }
        }
        
        Ok(VoteResponse {
            term: self.current_term,
            vote_granted,
        })
    }
    
    /// 处理投票响应
    pub async fn handle_vote_response(&mut self, response: VoteResponse) -> Result<()> {
        tracing::debug!("处理投票响应: {:?}", response);
        
        // 检查任期
        if response.term > self.current_term {
            self.current_term = response.term;
            self.voted_for = None;
            self.state = ConsensusState::Follower;
            return Ok(());
        }
        
        // 如果是候选者且获得投票
        if self.state == ConsensusState::Candidate && response.vote_granted {
            // 简化实现：假设获得足够投票就成为领导者
            self.become_leader().await?;
        }
        
        Ok(())
    }
    
    /// 成为领导者
    async fn become_leader(&mut self) -> Result<()> {
        tracing::info!("成为领导者，任期: {}", self.current_term);
        
        self.state = ConsensusState::Leader;
        
        // 发送领导者变更事件
        if let Some(sender) = &self.event_sender {
            let _ = sender.send(ClusterEvent::LeaderChanged(Some(self.config.node_id.clone())));
        }
        
        // 发送心跳
        self.send_heartbeat().await?;
        
        Ok(())
    }
    
    /// 发送心跳
    async fn send_heartbeat(&self) -> Result<()> {
        let append_request = AppendEntriesRequest {
            term: self.current_term,
            leader_id: self.config.node_id.clone(),
            prev_log_index: self.log.len() as u64,
            prev_log_term: self.log.last().map(|entry| entry.term).unwrap_or(0),
            entries: Vec::new(), // 心跳不包含日志条目
            leader_commit: self.commit_index,
        };
        
        tracing::debug!("发送心跳: {:?}", append_request);
        
        // 简化实现：实际需要向所有跟随者发送
        
        Ok(())
    }
    
    /// 处理追加条目请求
    pub async fn handle_append_entries(&mut self, request: AppendEntriesRequest) -> Result<AppendEntriesResponse> {
        tracing::debug!("处理追加条目请求: {:?}", request);
        
        let mut success = false;
        
        // 检查任期
        if request.term >= self.current_term {
            self.current_term = request.term;
            self.state = ConsensusState::Follower;
            self.reset_election_timeout();
            
            // 检查日志一致性
            if request.prev_log_index == 0 ||
               (request.prev_log_index <= self.log.len() as u64 &&
                self.log.get((request.prev_log_index - 1) as usize)
                    .map(|entry| entry.term) == Some(request.prev_log_term)) {
                
                success = true;
                
                // 追加新条目
                if !request.entries.is_empty() {
                    // 删除冲突的条目
                    self.log.truncate(request.prev_log_index as usize);
                    
                    // 追加新条目
                    self.log.extend(request.entries);
                }
                
                // 更新提交索引
                if request.leader_commit > self.commit_index {
                    self.commit_index = std::cmp::min(request.leader_commit, self.log.len() as u64);
                }
            }
        }
        
        Ok(AppendEntriesResponse {
            term: self.current_term,
            success,
        })
    }
    
    /// 追加日志条目
    pub async fn append_log(&mut self, command: Command) -> Result<u64> {
        if self.state != ConsensusState::Leader {
            return Err(Error::consensus("只有领导者可以追加日志".to_string()));
        }
        
        let entry = LogEntry {
            term: self.current_term,
            index: self.log.len() as u64 + 1,
            command,
            timestamp: SystemTime::now(),
        };
        
        self.log.push(entry);
        tracing::debug!("追加日志条目，索引: {}", self.log.len());
        
        Ok(self.log.len() as u64)
    }
    
    /// 重置选举超时
    fn reset_election_timeout(&mut self) {
        let timeout_ms = self.config.election_timeout_ms;
        self.election_timeout = SystemTime::now() + Duration::from_millis(timeout_ms);
    }
    
    /// 检查是否选举超时
    pub fn is_election_timeout(&self) -> bool {
        SystemTime::now() > self.election_timeout
    }
    
    /// 获取当前状态
    pub fn get_state(&self) -> ConsensusState {
        self.state.clone()
    }
    
    /// 获取当前任期
    pub fn get_current_term(&self) -> u64 {
        self.current_term
    }
    
    /// 设置事件发送器
    pub fn set_event_sender(&mut self, sender: mpsc::UnboundedSender<ClusterEvent>) {
        self.event_sender = Some(sender);
    }
}

impl crate::Error {
    /// 创建共识错误
    pub fn consensus<S: Into<String>>(msg: S) -> Self {
        Self::Storage(msg.into())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[tokio::test]
    async fn test_consensus_engine_creation() {
        let config = ClusterConfig::default();
        let consensus = ConsensusEngine::new(config).unwrap();
        
        assert_eq!(consensus.state, ConsensusState::Follower);
        assert_eq!(consensus.current_term, 0);
        assert_eq!(consensus.voted_for, None);
        assert_eq!(consensus.log.len(), 0);
    }
    
    #[tokio::test]
    async fn test_election_process() {
        let config = ClusterConfig::default();
        let mut consensus = ConsensusEngine::new(config).unwrap();
        
        // 开始选举
        consensus.start_election().await.unwrap();
        
        assert_eq!(consensus.state, ConsensusState::Candidate);
        assert_eq!(consensus.current_term, 1);
        assert_eq!(consensus.voted_for, Some("node-1".to_string()));
    }
    
    #[tokio::test]
    async fn test_vote_request_handling() {
        let config = ClusterConfig::default();
        let mut consensus = ConsensusEngine::new(config).unwrap();
        
        let vote_request = VoteRequest {
            term: 1,
            candidate_id: "candidate-1".to_string(),
            last_log_index: 0,
            last_log_term: 0,
        };
        
        let response = consensus.handle_vote_request(vote_request).await.unwrap();
        
        assert_eq!(response.term, 1);
        assert!(response.vote_granted);
        assert_eq!(consensus.voted_for, Some("candidate-1".to_string()));
    }
    
    #[tokio::test]
    async fn test_log_append() {
        let config = ClusterConfig::default();
        let mut consensus = ConsensusEngine::new(config).unwrap();
        
        // 成为领导者
        consensus.state = ConsensusState::Leader;
        consensus.current_term = 1;
        
        let command = Command::DataOperation {
            operation_type: "INSERT".to_string(),
            key: "test_key".to_string(),
            value: Some(b"test_value".to_vec()),
        };
        
        let index = consensus.append_log(command).await.unwrap();
        
        assert_eq!(index, 1);
        assert_eq!(consensus.log.len(), 1);
        assert_eq!(consensus.log[0].term, 1);
    }
}
